package com.wjk.kylin.rpc.core.invoker;

import com.wjk.kylin.rpc.core.registry.RegistryTypeEnum;
import com.wjk.kylin.rpc.core.invoker.future.RpcFutureResponse;
import com.wjk.kylin.rpc.core.netty.params.BaseCallback;
import com.wjk.kylin.rpc.core.netty.params.RpcResponse;
import com.wjk.kylin.rpc.core.registry.Register;
import com.wjk.kylin.rpc.core.util.RpcException;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * kylin-rpc consumer factory, init service-registry
 *
 * @author wjk
 */
public class RpcConsumerFactory {
    private static final Logger LOGGER = LoggerFactory.getLogger(RpcConsumerFactory.class);

    // ---------------------- config ----------------------
    /**
     * 注册中心类型
     */
    private final RegistryTypeEnum serviceRegistryType;
    /**
     * 注册中心地址
     */
    private final String serviceRegistryAddress;

    private Register register;
    private final List<BaseCallback> stopCallbackList = new ArrayList<>();

    public RpcConsumerFactory(RegistryTypeEnum serviceRegistryType, String serviceRegistryAddress) {
        this.serviceRegistryType = serviceRegistryType;
        this.serviceRegistryAddress = serviceRegistryAddress;
    }

    public Register getRegister() {
        return register;
    }

    public void addStopCallBack(BaseCallback callback) {
        stopCallbackList.add(callback);
    }

    // ---------------------- start / stop ----------------------

    public void start() throws Exception {
        // start registry
        if (this.serviceRegistryType == null) {
            throw new RpcException("kylin-rpc consumer init registry type is empty");
        }
        if (StringUtils.isBlank(this.serviceRegistryAddress)) {
            throw new RpcException("kylin-rpc consumer registry address missing");
        }
        Class<? extends Register> serviceRegistry = this.serviceRegistryType.getClazz();
        if (serviceRegistry == null) {
            throw new RpcException("kylin-rpc consumer registry missing");
        }
        //注册中心实例
        register = serviceRegistry.getDeclaredConstructor().newInstance();
        //初始化注册中心
        register.start(serviceRegistryAddress);
    }

    /**
     * 服务关闭
     *
     * @throws Exception
     */
    public void stop() throws Exception {
        //注册中心 关闭等处理
        if (register != null) {
            register.stop();
        }

        // stop callback
        if (stopCallbackList.size() > 0) {
            for (BaseCallback callback : stopCallbackList) {
                try {
                    callback.execute();
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                }
            }
        }
    }

    // ---------------------- future-response pool ----------------------

    /**
     * 请求对象集合
     */
    private final ConcurrentMap<String, RpcFutureResponse> futureResponsePool = new ConcurrentHashMap<>();

    public void putConsumerFuture(String requestId, RpcFutureResponse futureResponse) {
        futureResponsePool.put(requestId, futureResponse);
    }

    public void removeConsumerFuture(String requestId) {
        futureResponsePool.remove(requestId);
    }

    /**
     * 唤醒请求线程，返回执行结果
     *
     * @param requestId
     * @param rpcResponse
     */
    public void notifyConsumerFuture(String requestId, RpcResponse rpcResponse) {
        // get
        RpcFutureResponse futureResponse = futureResponsePool.get(requestId);
        if (futureResponse == null) {
            return;
        }

        // other nomal type
        futureResponse.setResponse(rpcResponse);

        // do remove
        removeConsumerFuture(requestId);
    }

}
